package com.telra.app.server.cassandra;

import com.telra.app.server.config.ConfigMgr;
import com.telra.app.server.data.OrderEvent;
import com.telra.app.server.protobuff.OrderProtos;
import com.telra.app.server.protobuff.OrderProtos.OrderTicket;
import me.prettyprint.cassandra.model.CqlQuery;
import me.prettyprint.cassandra.model.CqlRows;
import me.prettyprint.cassandra.serializers.StringSerializer;
import me.prettyprint.hector.api.Cluster;
import me.prettyprint.hector.api.Keyspace;
import me.prettyprint.hector.api.beans.ColumnSlice;
import me.prettyprint.hector.api.beans.HColumn;
import me.prettyprint.hector.api.beans.Row;
import me.prettyprint.hector.api.factory.HFactory;
import me.prettyprint.hector.api.query.QueryResult;
import org.apache.commons.logging.LogFactory;
import org.apache.commons.logging.impl.Log4JLogger;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.AnnotationConfigApplicationContext;

import java.text.DecimalFormat;
import java.util.ArrayList;
import java.util.List;

/**
 * Implementation of CQL to query cassandra for orders
 * Date: 1/4/12
 * Time: 6:40 PM
 */
public class CassandraCQLQuery {

    private final static Log4JLogger log = (Log4JLogger) LogFactory.getFactory().getInstance(CassandraCQLQuery.class);
    private final static DecimalFormat fmt = new DecimalFormat("#,##0.000000");

    private Keyspace keyspace;

    private CassandraCQLQuery() {
        Cluster cluster = HFactory.getOrCreateCluster("TestCluster", "localhost:9160");
        keyspace = HFactory.createKeyspace("OrderKeyspace", cluster);
    }

    private CqlQuery<String, String, String> getCqlQuery(boolean suppressKeyInColumns) {
        return new CqlQuery<String, String, String>(keyspace,
                StringSerializer.get(),   //key
                StringSerializer.get(),   //name
                StringSerializer.get()).setSuppressKeyInColumns(suppressKeyInColumns) ; //value
    }

    private OrderTicket.Category getCategory(String category) {
        if (category == null) throw new IllegalArgumentException("String cannot be null") ;

        if (category.equalsIgnoreCase("CLOTHING")) return OrderTicket.Category.CLOTHING;
        if (category.equalsIgnoreCase("MISC")) return OrderTicket.Category.MISC;
        if (category.equalsIgnoreCase("ACCESSORIES")) return OrderTicket.Category.ACCESSORIES;
        if (category.equalsIgnoreCase("FOOTWEAR")) return OrderTicket.Category.FOOTWEAR;
        if (category.equalsIgnoreCase("OUTERWEAR")) return OrderTicket.Category.OUTERWEAR;

        throw new IllegalArgumentException("Unknown category = " + category) ;
    }


    private List<OrderTicket> executeQuery(String query)  {
        return executeQuery(query, false);
    }

    private List<OrderTicket> executeQuery(String query, boolean suppressKeyInColumns)  {
        CqlQuery<String, String, String> cql = getCqlQuery(suppressKeyInColumns).setQuery(query);
        log.info("executing query=" + query );
        QueryResult<CqlRows<String, String, String>> cqlResult = cql.execute();
        log.info("done in " + (fmt.format(cqlResult.getExecutionTimeMicro()/1000000.0) ) );
        log.info("size = " + cqlResult.get().getCount() );

        List<OrderTicket> orders = new ArrayList<OrderTicket>();
        for (Row<String, String, String> row : cqlResult.get()) {

            //String key = row.getKey();
            ColumnSlice<String, String> colSlice = row.getColumnSlice();

            List<HColumn<String, String>> cols = colSlice.getColumns();

            OrderTicket.Category cat = getCategory(cols.get(0).getValue());
            
            OrderTicket order = OrderProtos.OrderTicket.newBuilder()
                    .setItemName(cols.get(1).getValue())
                    .setItemCategory(cat)
                    .setPrice(Double.valueOf(cols.get(2).getValue()))
                    .build();

            //log.info(order);
            orders.add(order);

            //for (HColumn col : cols) {
            //    log.info(col.getName() + "=" + col.getValue());
            //}
        }

        return orders;
    }
    
    public List<OrderTicket> orderQuery(String orderID) {
        return this.executeQuery("select category,itemName,price from Orders where KEY = '" + orderID + "'");

    }

    public List<OrderTicket> queryByCategory(OrderEvent.Category category, int rowlimit)  {
        return this.executeQuery("select category,itemName,price from Orders where category = '" + category + "' limit " + rowlimit, false);
    }

    public static CassandraCQLQuery getInstance()  {
        return new CassandraCQLQuery();
    }

    public static void main(String args[]) {

        ApplicationContext ctx = new AnnotationConfigApplicationContext(ConfigMgr.class);
        CassandraService cassandraService = ctx.getBean(CassandraService.class);
        cassandraService.getKeyspace();

        CassandraCQLQuery query = CassandraCQLQuery.getInstance();

        query.queryByCategory(OrderEvent.Category.FOOTWEAR, 1000);

        query.orderQuery("1326064730853");

    }
}


